package com.shengzai.stream

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo2UpdateStateByKey {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()

    conf.setMaster("local[2]")

    conf.setAppName("upstateByKey")

    val sc = new SparkContext(conf)

    val ssc = new StreamingContext(sc, Durations.seconds(8))

    //将之前的数据保存到data文件夹
    ssc.checkpoint("data/checkpoint")

    val sock: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.146.136", 8888)

    val tmpDS: DStream[(String, Int)] = sock.flatMap(_.split(",")).map((_, 1))

    def tm(seq:Seq[Int],option:Option[Int]):Option[Int]={

      val count: Int = seq.sum
      val i: Int = option.getOrElse(0)
      Option(count+i)
    }

    val res: DStream[(String, Int)] = tmpDS.updateStateByKey(tm)

    res.print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
